//! 行存储格式模块
//!
//! 实现高效的行存储格式，包括元组头部、NULL掩码、变长数据等
//! 支持MVCC版本控制和高效的数据访问

use crate::{Error, Result};
use crate::storage::page::{Page, PageId};
use crate::storage::Value; // 添加Value类型导入
use serde::{Serialize, Deserialize};
use std::collections::HashMap;

/// 行ID类型 - 由页面ID和槽位ID组成
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct RowId {
    /// 页面ID
    pub page_id: PageId,
    /// 槽位ID
    pub slot_id: u16,
}

/// 元组头部 - 16字节固定大小（支持MVCC）
#[repr(C)]
#[derive(Debug, Clone, Copy)]
pub struct TupleHeader {
    /// 元组大小（包括头部）
    pub size: u16,
    /// 标志位
    pub flags: TupleFlags,
    /// NULL掩码长度
    pub null_mask_len: u8,
    /// 列数量
    pub column_count: u8,
    /// 事务ID - 用于MVCC
    pub xmin: u32,
    /// 删除事务ID - 用于MVCC（0表示未删除）
    pub xmax: u32,
}

/// 元组标志位
#[derive(Debug, Clone, Copy)]
pub struct TupleFlags(pub u8);

impl TupleFlags {
    /// 元组有效
    pub const VALID: u8 = 0x01;
    /// 元组已删除
    pub const DELETED: u8 = 0x02;
    /// 元组已更新
    pub const UPDATED: u8 = 0x04;
    /// 元组有变长字段
    pub const HAS_VARLEN: u8 = 0x08;
    /// 元组有NULL值
    pub const HAS_NULL: u8 = 0x10;
    
    pub fn new() -> Self {
        Self(Self::VALID)
    }
    
    pub fn is_valid(&self) -> bool {
        self.0 & Self::VALID != 0
    }
    
    pub fn is_deleted(&self) -> bool {
        self.0 & Self::DELETED != 0
    }
    
    pub fn is_updated(&self) -> bool {
        self.0 & Self::UPDATED != 0
    }
    
    pub fn has_varlen(&self) -> bool {
        self.0 & Self::HAS_VARLEN != 0
    }
    
    pub fn has_null(&self) -> bool {
        self.0 & Self::HAS_NULL != 0
    }
    
    pub fn set_deleted(&mut self) {
        self.0 |= Self::DELETED;
        // 不要清除VALID标志，因为MVCC依赖事务ID来判断可见性
    }
    
    pub fn set_updated(&mut self) {
        self.0 |= Self::UPDATED;
    }
    
    pub fn set_has_varlen(&mut self) {
        self.0 |= Self::HAS_VARLEN;
    }
    
    pub fn set_has_null(&mut self) {
        self.0 |= Self::HAS_NULL;
    }
}



/// 元组数据结构
#[derive(Debug, Clone)]
pub struct Tuple {
    /// 元组头部
    pub header: TupleHeader,
    /// NULL掩码 - 每个位表示对应列是否为NULL
    pub null_mask: Vec<u8>,
    /// 列数据
    pub data: Vec<u8>,
}

/// 槽目录条目 - 4字节
#[repr(C)]
#[derive(Debug, Clone, Copy)]
pub struct SlotEntry {
    /// 元组在页面中的偏移量
    pub offset: u16,
    /// 元组长度
    pub length: u16,
}

/// 数据页布局管理器
/// 负责在页面中管理元组的存储和检索
#[derive(Debug)]
pub struct DataPageManager {
    /// 页面中的槽目录
    slot_directory: Vec<SlotEntry>,
    /// 下一个可用的槽位ID
    next_slot_id: u16,
    /// 空闲空间起始位置
    free_space_start: u16,
    /// 空闲空间结束位置
    free_space_end: u16,
    /// 元组数据存储（槽位ID -> 元组字节数据）
    /// 在真实实现中，这应该是页面缓冲区的一部分
    tuple_data: std::collections::HashMap<u16, Vec<u8>>,
}

impl RowId {
    /// 创建新的行ID
    pub fn new(page_id: PageId, slot_id: u16) -> Self {
        Self { page_id, slot_id }
    }
    
    /// 获取无效的行ID
    pub fn invalid() -> Self {
        Self {
            page_id: 0,
            slot_id: 0,
        }
    }
    
    /// 检查行ID是否有效
    pub fn is_valid(&self) -> bool {
        self.page_id != 0
    }
}

impl TupleHeader {
    /// 创建新的元组头部
    pub fn new(column_count: u8, xmin: u32) -> Self {
        Self {
            size: std::mem::size_of::<TupleHeader>() as u16,
            flags: TupleFlags::new(),
            null_mask_len: column_count.div_ceil(8), // 向上取整到字节
            column_count,
            xmin,
            xmax: 0, // 0表示未删除
        }
    }

    /// 获取创建事务ID
    pub fn xmin(&self) -> u32 {
        self.xmin
    }

    /// 获取删除事务ID
    pub fn xmax(&self) -> u32 {
        self.xmax
    }

    /// 设置删除事务ID
    pub fn set_xmax(&mut self, xmax: u32) {
        self.xmax = xmax;
        self.flags.set_deleted();
    }

    /// 检查元组是否已删除
    pub fn is_deleted(&self) -> bool {
        self.xmax > 0
    }
    
    /// 获取头部大小
    pub fn header_size() -> usize {
        std::mem::size_of::<TupleHeader>()
    }
    
    /// 标记为已删除
    pub fn mark_deleted(&mut self, xmax: u32) {
        self.flags.set_deleted();
        self.xmax = xmax;
        // 注意：不要设置flags为无效，因为MVCC依赖事务ID来判断可见性
    }
    
    /// 标记为已更新
    pub fn mark_updated(&mut self, xmax: u32) {
        self.flags.set_updated();
        self.xmax = xmax;
    }
    
    /// 检查元组对事务是否可见
    pub fn is_visible_to_transaction(&self, xid: u32) -> bool {
        // 简化的可见性检查
        // 实际实现需要考虑事务状态、快照等

        // 创建事务必须已提交且小于等于当前事务
        if self.xmin > xid {
            return false;
        }

        // 如果有删除事务，检查删除事务是否对当前事务可见
        // 如果删除事务ID小于当前事务ID，说明元组已被删除且对当前事务可见
        if self.xmax != 0 && self.xmax <= xid {
            return false;
        }

        true
    }
}

impl Tuple {
    /// 创建新元组
    pub fn new(column_count: u8, xmin: u32) -> Self {
        let header = TupleHeader::new(column_count, xmin);
        let null_mask = vec![0u8; header.null_mask_len as usize];
        
        Self {
            header,
            null_mask,
            data: Vec::new(),
        }
    }
    
    /// 从字节数据创建元组
    pub fn from_bytes(data: &[u8]) -> Result<Self> {
        if data.len() < TupleHeader::header_size() {
            return Err(Error::storage("元组数据太短".to_string()));
        }
        
        // 解析头部
        let header = unsafe {
            std::ptr::read(data.as_ptr() as *const TupleHeader)
        };
        
        if data.len() < header.size as usize {
            return Err(Error::storage("元组数据不完整".to_string()));
        }
        
        let mut offset = TupleHeader::header_size();
        
        // 解析NULL掩码
        let null_mask_len = header.null_mask_len as usize;
        let null_mask = data[offset..offset + null_mask_len].to_vec();
        offset += null_mask_len;
        
        // 解析数据
        let tuple_data = data[offset..header.size as usize].to_vec();
        
        Ok(Self {
            header,
            null_mask,
            data: tuple_data,
        })
    }
    
    /// 将元组序列化为字节数据
    pub fn to_bytes(&self) -> Vec<u8> {
        let mut bytes = Vec::new();
        
        // 序列化头部
        let header_bytes = unsafe {
            std::slice::from_raw_parts(
                &self.header as *const TupleHeader as *const u8,
                TupleHeader::header_size(),
            )
        };
        bytes.extend_from_slice(header_bytes);
        
        // 序列化NULL掩码
        bytes.extend_from_slice(&self.null_mask);
        
        // 序列化数据
        bytes.extend_from_slice(&self.data);

        bytes
    }

    /// 从值向量创建元组
    pub fn from_values(values: Vec<Value>, xmin: u32, xmax: u32) -> Result<Self> {
        let column_count = values.len() as u8;
        let mut header = TupleHeader::new(column_count, xmin);
        header.xmax = xmax;

        // 计算NULL掩码长度
        let null_mask_len = column_count.div_ceil(8);
        let mut null_mask = vec![0u8; null_mask_len as usize];

        // 序列化数据
        let mut data = Vec::new();
        for (i, value) in values.iter().enumerate() {
            match value {
                Value::Null => {
                    // 设置NULL位
                    let byte_idx = i / 8;
                    let bit_idx = i % 8;
                    null_mask[byte_idx] |= 1 << bit_idx;
                    header.flags.set_has_null();
                }
                Value::Boolean(b) => {
                    data.push(if *b { 1 } else { 0 });
                }
                Value::Integer(i) => {
                    data.extend_from_slice(&i.to_le_bytes());
                }
                Value::Real(f) => {
                    data.extend_from_slice(&f.to_le_bytes());
                }
                Value::Text(s) => {
                    let bytes = s.as_bytes();
                    data.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
                    data.extend_from_slice(bytes);
                    header.flags.set_has_varlen();
                }
                Value::Timestamp(ts) => {
                    // 将时间戳转换为字符串存储
                    let ts_str = ts.to_rfc3339();
                    let bytes = ts_str.as_bytes();
                    data.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
                    data.extend_from_slice(bytes);
                    header.flags.set_has_varlen();
                }
                Value::Decimal(d) => {
                    // 将Decimal转换为字符串存储
                    let decimal_str = d.to_string();
                    let bytes = decimal_str.as_bytes();
                    data.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
                    data.extend_from_slice(bytes);
                    header.flags.set_has_varlen();
                }
            }
        }

        // 更新头部大小
        header.size = (TupleHeader::header_size() + null_mask_len as usize + data.len()) as u16;
        header.null_mask_len = null_mask_len;

        Ok(Self {
            header,
            null_mask,
            data,
        })
    }

    /// 将元组转换为值向量
    pub fn to_values(&self, schema: &crate::storage::TableSchema) -> Result<Vec<Value>> {
        let mut values = Vec::new();
        let mut data_offset = 0;

        for (i, column) in schema.columns.iter().enumerate() {
            // 检查是否为NULL
            let byte_idx = i / 8;
            let bit_idx = i % 8;
            if byte_idx < self.null_mask.len() && (self.null_mask[byte_idx] & (1 << bit_idx)) != 0 {
                values.push(Value::Null);
                continue;
            }

            // 根据列类型解析数据
            let value = match &column.data_type {
                crate::storage::DataType::Boolean => {
                    if data_offset >= self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let val = self.data[data_offset] != 0;
                    data_offset += 1;
                    Value::Boolean(val)
                }
                crate::storage::DataType::Integer => {
                    if data_offset + 8 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let bytes = &self.data[data_offset..data_offset + 8];
                    let val = i64::from_le_bytes(bytes.try_into().unwrap());
                    data_offset += 8;
                    Value::Integer(val)
                }
                crate::storage::DataType::Real => {
                    if data_offset + 8 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let bytes = &self.data[data_offset..data_offset + 8];
                    let val = f64::from_le_bytes(bytes.try_into().unwrap());
                    data_offset += 8;
                    Value::Real(val)
                }
                crate::storage::DataType::Text(_) => {
                    if data_offset + 4 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let len_bytes = &self.data[data_offset..data_offset + 4];
                    let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
                    data_offset += 4;

                    if data_offset + len > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let text_bytes = &self.data[data_offset..data_offset + len];
                    let val = String::from_utf8(text_bytes.to_vec())
                        .map_err(|_| Error::storage("无效的UTF-8文本".to_string()))?;
                    data_offset += len;
                    Value::Text(val)
                }
                crate::storage::DataType::BigInt => {
                    if data_offset + 8 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let bytes = &self.data[data_offset..data_offset + 8];
                    let val = i64::from_le_bytes(bytes.try_into().unwrap());
                    data_offset += 8;
                    Value::Integer(val)
                }
                crate::storage::DataType::Double => {
                    if data_offset + 8 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let bytes = &self.data[data_offset..data_offset + 8];
                    let val = f64::from_le_bytes(bytes.try_into().unwrap());
                    data_offset += 8;
                    Value::Real(val)
                }
                crate::storage::DataType::Date => {
                    // 简化实现：日期作为文本存储
                    if data_offset + 4 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let len_bytes = &self.data[data_offset..data_offset + 4];
                    let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
                    data_offset += 4;

                    if data_offset + len > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let text_bytes = &self.data[data_offset..data_offset + len];
                    let val = String::from_utf8(text_bytes.to_vec())
                        .map_err(|_| Error::storage("无效的UTF-8文本".to_string()))?;
                    data_offset += len;
                    Value::Text(val)
                }
                crate::storage::DataType::Timestamp => {
                    // 简化实现：时间戳作为文本存储
                    if data_offset + 4 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let len_bytes = &self.data[data_offset..data_offset + 4];
                    let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
                    data_offset += 4;

                    if data_offset + len > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let text_bytes = &self.data[data_offset..data_offset + len];
                    let val = String::from_utf8(text_bytes.to_vec())
                        .map_err(|_| Error::storage("无效的UTF-8文本".to_string()))?;
                    data_offset += len;
                    Value::Text(val)
                }
                crate::storage::DataType::Decimal(_) => {
                    // Decimal作为文本存储
                    if data_offset + 4 > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let len_bytes = &self.data[data_offset..data_offset + 4];
                    let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
                    data_offset += 4;

                    if data_offset + len > self.data.len() {
                        return Err(Error::storage("元组数据不足".to_string()));
                    }
                    let text_bytes = &self.data[data_offset..data_offset + len];
                    let decimal_str = String::from_utf8(text_bytes.to_vec())
                        .map_err(|_| Error::storage("无效的UTF-8文本".to_string()))?;
                    data_offset += len;

                    let decimal = decimal_str.parse::<rust_decimal::Decimal>()
                        .map_err(|_| Error::storage("无效的Decimal格式".to_string()))?;
                    Value::Decimal(decimal)
                }
            };

            values.push(value);
        }

        Ok(values)
    }
    
    /// 设置列为NULL
    pub fn set_null(&mut self, column_index: usize) -> Result<()> {
        if column_index >= self.header.column_count as usize {
            return Err(Error::storage("列索引超出范围".to_string()));
        }
        
        let byte_index = column_index / 8;
        let bit_index = column_index % 8;
        
        if byte_index >= self.null_mask.len() {
            return Err(Error::storage("NULL掩码索引超出范围".to_string()));
        }
        
        self.null_mask[byte_index] |= 1 << bit_index;
        self.header.flags.set_has_null();
        
        Ok(())
    }
    
    /// 检查列是否为NULL
    pub fn is_null(&self, column_index: usize) -> Result<bool> {
        if column_index >= self.header.column_count as usize {
            return Err(Error::storage("列索引超出范围".to_string()));
        }
        
        let byte_index = column_index / 8;
        let bit_index = column_index % 8;
        
        if byte_index >= self.null_mask.len() {
            return Ok(false);
        }
        
        Ok(self.null_mask[byte_index] & (1 << bit_index) != 0)
    }
    
    /// 添加列数据
    pub fn add_column_data(&mut self, data: &[u8]) {
        self.data.extend_from_slice(data);
        self.header.size = (TupleHeader::header_size() + self.null_mask.len() + self.data.len()) as u16;
    }
    
    /// 获取元组总大小
    pub fn total_size(&self) -> usize {
        TupleHeader::header_size() + self.null_mask.len() + self.data.len()
    }
    
    /// 检查元组对事务是否可见
    pub fn is_visible_to_transaction(&self, xid: u32) -> bool {
        self.header.is_visible_to_transaction(xid)
    }
    
    /// 标记为已删除
    pub fn mark_deleted(&mut self, xmax: u32) {
        self.header.mark_deleted(xmax);
    }
    
    /// 标记为已更新
    pub fn mark_updated(&mut self, xmax: u32) {
        self.header.mark_updated(xmax);
    }
}

impl SlotEntry {
    /// 创建新的槽目录条目
    pub fn new(offset: u16, length: u16) -> Self {
        Self { offset, length }
    }
    
    /// 获取无效的槽条目
    pub fn invalid() -> Self {
        Self {
            offset: 0,
            length: 0,
        }
    }
    
    /// 检查槽条目是否有效
    pub fn is_valid(&self) -> bool {
        self.length > 0
    }
}

impl DataPageManager {
    /// 创建新的数据页管理器
    pub fn new() -> Self {
        Self {
            slot_directory: Vec::new(),
            next_slot_id: 0,
            // 页面数据区域从0开始，因为Page.data已经不包含PageHeader
            free_space_start: 0,
            // 页面数据区域的大小是PAGE_SIZE减去PageHeader的大小
            free_space_end: (crate::storage::page::PAGE_SIZE - std::mem::size_of::<crate::storage::page::PageHeader>()) as u16,
            // 初始化元组数据存储
            tuple_data: std::collections::HashMap::new(),
        }
    }
    

    

    
    /// 删除页面中的元组
    pub fn delete_tuple(&mut self, page: &mut Page, slot_id: u16) -> Result<()> {
        if slot_id as usize >= self.slot_directory.len() {
            return Err(Error::storage("槽位ID无效".to_string()));
        }
        
        // 标记槽位为无效
        self.slot_directory[slot_id as usize] = SlotEntry::invalid();
        
        // 更新页面头部
        page.header.slot_count -= 1;
        page.mark_dirty();
        
        // TODO: 实现空间回收和碎片整理
        
        Ok(())
    }
    
    /// 获取可用空间大小
    pub fn available_space(&self) -> usize {
        if self.free_space_end > self.free_space_start {
            (self.free_space_end - self.free_space_start) as usize
        } else {
            0
        }
    }
    
    /// 获取页面中的元组数量
    pub fn tuple_count(&self) -> usize {
        self.slot_directory.iter().filter(|slot| slot.is_valid()).count()
    }

    /// 插入元组到指定页面（真实实现）
    pub fn insert_tuple(&mut self, page_id: PageId, tuple: &Tuple) -> Result<u16> {
        let slot_id = self.next_slot_id;
        self.next_slot_id += 1;

        // 序列化元组数据
        let tuple_bytes = tuple.to_bytes();
        let tuple_size = tuple_bytes.len() as u16;

        // 检查页面空间是否足够
        let required_space = tuple_size + std::mem::size_of::<SlotEntry>() as u16;
        if self.free_space_end < self.free_space_start + required_space {
            return Err(Error::storage("页面空间不足".to_string()));
        }

        // 计算元组在页面中的偏移位置
        let tuple_offset = self.free_space_end - tuple_size;

        // 创建槽目录条目
        let slot_entry = SlotEntry::new(tuple_offset, tuple_size);
        self.slot_directory.push(slot_entry);

        // 将元组数据存储到内存中（模拟页面数据）
        // 在真实实现中，这里应该写入到实际的页面缓冲区
        self.tuple_data.insert(slot_id, tuple_bytes);

        // 更新空闲空间指针
        self.free_space_end = tuple_offset;
        self.free_space_start += std::mem::size_of::<SlotEntry>() as u16;

        tracing::debug!("插入元组到页面 {} 槽位 {}，偏移 {}，大小 {}",
                       page_id, slot_id, tuple_offset, tuple_size);
        Ok(slot_id)
    }

    /// 从指定页面读取元组（兼容Table接口）
    pub fn read_tuple(&self, page_id: PageId, slot_id: u16, page: &crate::storage::page::Page) -> Result<Option<Tuple>> {
        if slot_id as usize >= self.slot_directory.len() {
            return Ok(None);
        }

        let slot_entry = &self.slot_directory[slot_id as usize];
        if !slot_entry.is_valid() {
            return Ok(None);
        }

        // 优先从tuple_data中读取更新后的元组数据
        if let Some(updated_data) = self.tuple_data.get(&slot_id) {
            tracing::debug!("从tuple_data中读取更新后的元组，slot_id: {}, 数据长度: {}", slot_id, updated_data.len());
            match Tuple::from_bytes(updated_data) {
                Ok(tuple) => return Ok(Some(tuple)),
                Err(e) => {
                    tracing::warn!("解析更新后的元组数据失败: {}", e);
                    // 继续尝试从页面数据中读取
                }
            }
        }

        // 从页面数据中读取原始元组
        let page_data = &page.data;
        if slot_entry.offset as usize + slot_entry.length as usize > page_data.len() {
            return Ok(None);
        }

        let tuple_data = &page_data[slot_entry.offset as usize..
            (slot_entry.offset + slot_entry.length) as usize];

        match Tuple::from_bytes(tuple_data) {
            Ok(tuple) => Ok(Some(tuple)),
            Err(_) => Ok(None),
        }
    }

    /// 更新指定页面的元组（兼容Table接口）
    pub fn update_tuple(&mut self, page_id: PageId, slot_id: u16, tuple: &Tuple) -> Result<()> {
        if slot_id as usize >= self.slot_directory.len() {
            return Err(Error::storage("槽位ID无效".to_string()));
        }

        // 序列化新的元组数据
        let tuple_bytes = tuple.to_bytes();

        // 验证元组数据的完整性
        if tuple_bytes.len() < TupleHeader::header_size() {
            return Err(Error::storage("元组数据不完整".to_string()));
        }

        // 更新槽目录中的元组信息
        let slot_entry = &mut self.slot_directory[slot_id as usize];
        slot_entry.length = tuple_bytes.len() as u16;

        // 将新的元组数据存储到内存中（简化实现）
        // 在真实实现中，这里应该将数据写入到页面的具体位置
        // 目前我们将数据存储在一个HashMap中，以slot_id为键
        let tuple_len = tuple_bytes.len();
        self.tuple_data.insert(slot_id, tuple_bytes);

        tracing::debug!("更新页面 {} 槽位 {} 的元组，新长度: {}", page_id, slot_id, tuple_len);
        Ok(())
    }

    /// 扫描页面中的所有元组（兼容Table接口）
    pub fn scan_tuples(&self, _page_id: PageId, _page: &crate::storage::page::Page) -> Result<Vec<Tuple>> {
        let mut tuples = Vec::new();

        for (slot_id, slot_entry) in self.slot_directory.iter().enumerate() {
            if !slot_entry.is_valid() {
                continue;
            }

            // 从内存中的tuple_data读取元组数据
            if let Some(tuple_bytes) = self.tuple_data.get(&(slot_id as u16)) {
                if let Ok(tuple) = Tuple::from_bytes(tuple_bytes) {
                    tuples.push(tuple);
                }
            }
        }

        tracing::debug!("扫描到 {} 个元组", tuples.len());
        Ok(tuples)
    }

    /// 扫描页面中的所有元组并返回槽位ID（用于UPDATE/DELETE操作）
    pub fn scan_tuples_with_slot_id(&self, _page_id: PageId, _page: &crate::storage::page::Page) -> Result<Vec<(u16, Tuple)>> {
        let mut tuples_with_slots = Vec::new();

        for (slot_id, slot_entry) in self.slot_directory.iter().enumerate() {
            if !slot_entry.is_valid() {
                continue;
            }

            // 从内存中的tuple_data读取元组数据
            if let Some(tuple_bytes) = self.tuple_data.get(&(slot_id as u16)) {
                if let Ok(tuple) = Tuple::from_bytes(tuple_bytes) {
                    tuples_with_slots.push((slot_id as u16, tuple));
                }
            }
        }

        tracing::debug!("扫描到 {} 个元组（带槽位ID）", tuples_with_slots.len());
        Ok(tuples_with_slots)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::page::{PageType, PAGE_SIZE};

    #[test]
    fn test_row_id() {
        let row_id = RowId::new(1, 5);
        assert_eq!(row_id.page_id, 1);
        assert_eq!(row_id.slot_id, 5);
        assert!(row_id.is_valid());

        let invalid_row_id = RowId::invalid();
        assert!(!invalid_row_id.is_valid());
    }

    #[test]
    fn test_tuple_flags() {
        let mut flags = TupleFlags::new();
        assert!(flags.is_valid());
        assert!(!flags.is_deleted());

        flags.set_deleted();
        assert!(flags.is_valid()); // 修改：删除后仍然有效，因为MVCC依赖事务ID
        assert!(flags.is_deleted());

        flags.set_has_null();
        assert!(flags.has_null());

        flags.set_has_varlen();
        assert!(flags.has_varlen());
    }

    #[test]
    fn test_tuple_header() {
        let mut header = TupleHeader::new(5, 100);
        assert_eq!(header.column_count, 5);
        assert_eq!(header.xmin, 100);
        assert_eq!(header.xmax, 0);
        assert!(header.is_visible_to_transaction(100));
        assert!(header.is_visible_to_transaction(101));
        assert!(!header.is_visible_to_transaction(99));

        header.mark_deleted(200);
        assert!(!header.is_visible_to_transaction(200));
        assert!(!header.is_visible_to_transaction(201));
    }

    #[test]
    fn test_tuple_creation() {
        let tuple = Tuple::new(3, 100);
        assert_eq!(tuple.header.column_count, 3);
        assert_eq!(tuple.header.xmin, 100);
        assert_eq!(tuple.null_mask.len(), 1); // (3 + 7) / 8 = 1
    }

    #[test]
    fn test_tuple_null_handling() {
        let mut tuple = Tuple::new(8, 100);

        // 设置第0列为NULL
        tuple.set_null(0).unwrap();
        assert!(tuple.is_null(0).unwrap());
        assert!(!tuple.is_null(1).unwrap());

        // 设置第7列为NULL
        tuple.set_null(7).unwrap();
        assert!(tuple.is_null(7).unwrap());

        // 测试超出范围的列
        assert!(tuple.set_null(8).is_err());
        assert!(tuple.is_null(8).is_err());
    }

    #[test]
    fn test_tuple_serialization() {
        let mut tuple = Tuple::new(2, 100);
        tuple.set_null(0).unwrap();
        tuple.add_column_data(b"test_data");

        let bytes = tuple.to_bytes();
        let deserialized = Tuple::from_bytes(&bytes).unwrap();

        assert_eq!(deserialized.header.column_count, tuple.header.column_count);
        assert_eq!(deserialized.header.xmin, tuple.header.xmin);
        assert_eq!(deserialized.null_mask, tuple.null_mask);
        assert_eq!(deserialized.data, tuple.data);
        assert!(deserialized.is_null(0).unwrap());
    }

    #[test]
    fn test_slot_entry() {
        let slot = SlotEntry::new(100, 50);
        assert_eq!(slot.offset, 100);
        assert_eq!(slot.length, 50);
        assert!(slot.is_valid());

        let invalid_slot = SlotEntry::invalid();
        assert!(!invalid_slot.is_valid());
    }

    #[test]
    fn test_data_page_manager() {
        let mut page_manager = DataPageManager::new();
        let mut page = Page::new(1, PageType::Data);

        // 创建测试元组
        let mut tuple = Tuple::new(2, 100);
        tuple.add_column_data(b"hello");
        tuple.add_column_data(b"world");

        // 插入元组
        let slot_id = page_manager.insert_tuple(1, &tuple).unwrap();
        assert_eq!(slot_id, 0);
        assert_eq!(page_manager.tuple_count(), 1);

        // 读取元组
        let read_tuple = page_manager.read_tuple(1, slot_id, &page).unwrap().unwrap();
        assert_eq!(read_tuple.header.column_count, tuple.header.column_count);
        assert_eq!(read_tuple.data, tuple.data);

        // 删除元组
        page_manager.delete_tuple(&mut page, slot_id).unwrap();
        assert_eq!(page_manager.tuple_count(), 0);

        // 尝试读取已删除的元组
        assert!(page_manager.read_tuple(1, slot_id, &page).unwrap().is_none());
    }

    #[test]
    fn test_page_space_management() {
        let mut page_manager = DataPageManager::new();
        let _page = Page::new(1, PageType::Data);

        let initial_space = page_manager.available_space();
        assert!(initial_space > 0);

        // 插入一个元组
        let mut tuple = Tuple::new(1, 100);
        tuple.add_column_data(b"test");

        let tuple_size = tuple.total_size();
        page_manager.insert_tuple(1, &tuple).unwrap();

        // 检查空间减少
        let remaining_space = page_manager.available_space();
        assert!(remaining_space < initial_space);

        // 空间减少应该等于元组大小加上槽目录条目大小
        let expected_reduction = tuple_size + std::mem::size_of::<SlotEntry>();
        assert_eq!(initial_space - remaining_space, expected_reduction);
    }

    #[test]
    fn test_mvcc_visibility() {
        let mut tuple = Tuple::new(1, 100);

        // 事务100创建的元组
        assert!(tuple.is_visible_to_transaction(100));
        assert!(tuple.is_visible_to_transaction(101));
        assert!(!tuple.is_visible_to_transaction(99));

        // 事务200删除元组
        tuple.mark_deleted(200);
        assert!(tuple.is_visible_to_transaction(199));
        assert!(!tuple.is_visible_to_transaction(200));
        assert!(!tuple.is_visible_to_transaction(201));
    }

    #[test]
    fn test_tuple_with_real_values() {
        use crate::sql::Value;

        // 创建包含真实Value数据的元组
        let values = vec![
            Value::Integer(42),
            Value::Text("Hello, CDB!".to_string()),
            Value::Real(3.14159),
            Value::Boolean(true),
        ];

        let mut tuple = Tuple::new(values.len() as u8, 100);

        // 序列化Value数据
        for value in &values {
            match value {
                Value::Integer(i) => tuple.add_column_data(&i.to_le_bytes()),
                Value::Text(s) => {
                    // 先写入长度，再写入字符串数据
                    tuple.add_column_data(&(s.len() as u32).to_le_bytes());
                    tuple.add_column_data(s.as_bytes());
                },
                Value::Decimal(d) => {
                    let decimal_str = d.to_string();
                    let decimal_bytes = decimal_str.as_bytes();
                    buffer.extend_from_slice(&(decimal_bytes.len() as u32).to_le_bytes());
                    buffer.extend_from_slice(decimal_bytes);
                },
                Value::Real(f) => tuple.add_column_data(&f.to_le_bytes()),
                Value::Boolean(b) => tuple.add_column_data(&[*b as u8]),
                _ => {} // 简化处理其他类型
            }
        }

        // 测试序列化和反序列化
        let bytes = tuple.to_bytes();
        let deserialized = Tuple::from_bytes(&bytes).unwrap();

        assert_eq!(deserialized.header.column_count, values.len() as u8);
        assert_eq!(deserialized.data, tuple.data);
        assert_eq!(deserialized.header.xmin, 100);
    }

    #[test]
    fn test_tuple_page_integration() {
        use crate::storage::page::PageManager;
        use std::sync::Arc;

        // 创建页面管理器
        let page_manager = Arc::new(PageManager::new(10));
        let page_id = page_manager.allocate_page(PageType::Data).unwrap();

        // 获取页面 - 简化处理，直接创建页面
        let mut page = Page::new(page_id, PageType::Data);

        // 创建数据页管理器
        let mut data_page_manager = DataPageManager::new();

        // 创建多个元组并插入
        let mut slot_ids = Vec::new();
        for i in 0..5 {
            let mut tuple = Tuple::new(2, 100 + i);
            tuple.add_column_data(&(i as u32).to_le_bytes());
            tuple.add_column_data(format!("tuple_{}", i).as_bytes());

            let slot_id = data_page_manager.insert_tuple(1, &tuple).unwrap();
            slot_ids.push(slot_id);
        }

        // 验证所有元组都能正确读取
        for (i, slot_id) in slot_ids.iter().enumerate() {
            let tuple = data_page_manager.read_tuple(1, *slot_id, &page).unwrap().unwrap();
            assert_eq!(tuple.header.xmin, 100 + i as u32);
            assert_eq!(tuple.header.column_count, 2);
        }

        // 测试元组删除
        data_page_manager.delete_tuple(&mut page, slot_ids[2]).unwrap();
        assert!(data_page_manager.read_tuple(1, slot_ids[2], &page).unwrap().is_none());

        // 其他元组应该仍然可读
        assert!(data_page_manager.read_tuple(1, slot_ids[0], &page).is_ok());
        assert!(data_page_manager.read_tuple(1, slot_ids[4], &page).is_ok());
    }

    #[test]
    fn test_tuple_null_mask_edge_cases() {
        // 测试边界情况：8列（正好1字节）
        let mut tuple8 = Tuple::new(8, 100);
        assert_eq!(tuple8.null_mask.len(), 1);

        // 设置所有列为NULL
        for i in 0..8 {
            tuple8.set_null(i).unwrap();
            assert!(tuple8.is_null(i).unwrap());
        }

        // 测试边界情况：9列（需要2字节）
        let mut tuple9 = Tuple::new(9, 100);
        assert_eq!(tuple9.null_mask.len(), 2);

        // 设置第9列（索引8）为NULL
        tuple9.set_null(8).unwrap();
        assert!(tuple9.is_null(8).unwrap());

        // 前8列应该不是NULL
        for i in 0..8 {
            assert!(!tuple9.is_null(i).unwrap());
        }
    }

    #[test]
    fn test_tuple_size_calculations() {
        let tuple = Tuple::new(3, 100);

        // 基础大小：头部 + NULL掩码
        let base_size = std::mem::size_of::<TupleHeader>() + tuple.null_mask.len();
        assert_eq!(tuple.total_size(), base_size);

        // 添加数据后的大小
        let mut tuple_with_data = Tuple::new(3, 100);
        tuple_with_data.add_column_data(b"test_data");

        let expected_size = base_size + b"test_data".len();
        assert_eq!(tuple_with_data.total_size(), expected_size);
    }

    #[test]
    fn test_concurrent_tuple_operations() {
        use std::thread;
        use std::sync::{Arc, Mutex};

        let page_manager = Arc::new(Mutex::new(DataPageManager::new()));
        let page = Arc::new(Mutex::new(Page::new(1, PageType::Data)));
        let mut handles = vec![];

        // 启动多个线程并发插入元组
        for i in 0..5 {
            let page_manager_clone = Arc::clone(&page_manager);
            let page_clone = Arc::clone(&page);

            let handle = thread::spawn(move || {
                let mut tuple = Tuple::new(1, 100 + i);
                tuple.add_column_data(&(i as u32).to_le_bytes());

                let _page_guard = page_clone.lock().unwrap();
                let mut page_manager_guard = page_manager_clone.lock().unwrap();
                page_manager_guard.insert_tuple(1, &tuple).unwrap()
            });

            handles.push(handle);
        }

        // 等待所有线程完成
        let slot_ids: Vec<_> = handles.into_iter()
            .map(|h| h.join().unwrap())
            .collect();

        // 验证所有元组都被正确插入
        let page_guard = page.lock().unwrap();
        let page_manager_guard = page_manager.lock().unwrap();
        for slot_id in slot_ids {
            let tuple = page_manager_guard.read_tuple(1, slot_id, &*page_guard).unwrap().unwrap();
            assert!(tuple.header.xmin >= 100 && tuple.header.xmin < 105);
        }
    }
}
